package com.xg.mina;

import com.google.common.collect.Lists;
import com.xg.mina.filter.ShakeFilter;
import com.xg.mina.message.BasicHeader;
import com.xg.mina.message.Constant;
import com.xg.mina.message.MessageHeader;
import com.xg.mina.message.RtmpHeader;
import org.apache.commons.codec.binary.Hex;
import org.apache.commons.lang3.StringUtils;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.service.IoConnector;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketConnector;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

public class MinaTcpClient extends IoHandlerAdapter {
    public static final String SEND_INDEX = "send_index";
    public static final String SEND_MSG_2 = "send_msg_2";
    public static final String SEND_MSG_3 = "send_msg_3";

    public static final String ACK_INDEX = "ack_index";
    public static final String ACK_MSG2 = "act_msg2";
    public static final String ACK_MSG3 = "act_msg3";

    private static final AtomicInteger sendMessage = new AtomicInteger();

    private IoConnector connector;
    private static IoSession session;

    public MinaTcpClient() {
        String[] split = Utils.rtmp_url.split(":");
        String host = split[1].substring(2);
        Integer port = Integer.valueOf(StringUtils.substringBefore(split[2], "/"));

        connector = new NioSocketConnector();
        connector.setHandler(this);

        connector.getFilterChain().addFirst("shake", new ShakeFilter());
        connector.getFilterChain().addLast("connect", new ProtocolCodecFilter(new ConnectCodecFactory()));
        connector.getFilterChain().addLast("logger", new LoggingFilter());

        ConnectFuture connFuture = connector.connect(new InetSocketAddress(host, port));
        connFuture.awaitUninterruptibly();
        session = connFuture.getSession();
    }

    @Override
    public void sessionOpened(IoSession session) throws Exception {

    }

    @Override
    public void messageReceived(IoSession iosession, Object message) throws Exception {
        if (message instanceof ShakeMessage) {
            ShakeMessage recMessage = (ShakeMessage) message;
            Integer ack = (Integer) session.getAttribute(MinaTcpClient.ACK_INDEX);
            if (ack == null) {
                return;
            }
            if (ack <= 2) {
                byte[] data = recMessage.getData();
                if (data.length == 1) {
                    System.out.println("收到S0：" + recMessage);
                } else {
                    System.out.println("收到S1：" + recMessage);
                    session.setAttribute(MinaTcpClient.SEND_INDEX, 3);
                }
            } else if (ack == 3) {
                System.out.println("收到S2：" + recMessage);
                Thread.sleep(2000);
                //connection(iosession);
            }
        } else {
            System.out.println(message);
        }
    }

    public static Integer getChunkStreamId(IoSession session) {
        Integer chunk_stream_id = (Integer) session.getAttribute(Constant.chunk_stream_key);
        if (chunk_stream_id == null) {
            //从3开始，使用一个字节
            chunk_stream_id = Constant.chunk_stream_start;
        } else {
            chunk_stream_id++;
        }
        session.setAttribute(Constant.chunk_stream_key, chunk_stream_id);
        return chunk_stream_id;
    }

    public static int getTimestamp(IoSession session) {
        Long current = System.currentTimeMillis();
        //开始时间
        Long start_timestamp = (Long) session.getAttribute(Constant.timestamp_start);
        if (start_timestamp == null) {
            start_timestamp = current;
            session.setAttribute(Constant.timestamp_start, current);
        }
        System.out.println("相差时间:" + (current - start_timestamp));
        Long offset = current - start_timestamp;
        return offset.intValue();
    }

    public static void cleanTimestamp(IoSession session) {
        session.setAttribute(Constant.timestamp_key, null);
    }

    public static Integer getMessageStreamId(IoSession session) {
        Integer messageStreamId = (Integer) session.getAttribute(Constant.message_stream_key);
        if (messageStreamId == null) {
            //从3开始，使用一个字节
            messageStreamId = 0;
        } else {
            messageStreamId++;
        }
        session.setAttribute(Constant.message_stream_key, messageStreamId);
        return messageStreamId;
    }

    @Override
    public void exceptionCaught(IoSession session, Throwable cause)
            throws Exception {
        System.out.println("客户端异常");
        cause.printStackTrace();
        super.exceptionCaught(session, cause);
        throw new RuntimeException(cause.getMessage());
    }

    public static String byteToBit(byte b) {
        return ""
                + (byte) ((b >> 7) & 0x1) + (byte) ((b >> 6) & 0x1)
                + (byte) ((b >> 5) & 0x1) + (byte) ((b >> 4) & 0x1)
                + (byte) ((b >> 3) & 0x1) + (byte) ((b >> 2) & 0x1)
                + (byte) ((b >> 1) & 0x1) + (byte) ((b >> 0) & 0x1);
    }

    @Override
    public void messageSent(IoSession iosession, Object obj) throws Exception {
        System.out.println("已发送消息消息+======" + sendMessage.getAndIncrement());
        if (obj instanceof ConnectCommand) {
            ConnectCommand connectCommand = (ConnectCommand) obj;
            byte[] bytes = connectCommand.getBytes();
            StringBuilder sb = new StringBuilder();
            for (byte b : bytes) {
                sb.append(byteToBit(b)+" \t");
            }
            System.out.println("connect发送成功：" + sb);
            System.out.println(Hex.encodeHexString(bytes));
        } else {
            Integer sendIndex = (Integer) session.getAttribute(MinaTcpClient.SEND_INDEX);
            if (sendIndex == 2) {
                byte[] c1 = (byte[]) session.getAttribute(MinaTcpClient.SEND_MSG_2);
                System.out.println("c0c1发送成功，C1:" + Hex.encodeHexString(c1));
            } else if (sendIndex == 3) {
                byte[] c2 = (byte[]) session.getAttribute(MinaTcpClient.SEND_MSG_3);
                System.out.println("c2发送成功，C2:" + Hex.encodeHexString(c2));
            }
        }
    }

    public static void connection(IoSession iosession) {
        System.out.println("发送connect请求");
        ConnectCommand connectCommand = new ConnectCommand();
        RtmpHeader rtmpHeader = new RtmpHeader();

        Integer chunk_stream_id = getChunkStreamId(iosession);
        BasicHeader basicHeader = new BasicHeader(chunk_stream_id);
        //消息首次发送从00开始，11个字节
        basicHeader.setFmt(Constant.fmt00_11);
        rtmpHeader.setBasicHeader(basicHeader);

        connectCommand.setRtmpHeader(rtmpHeader);
        ConnectCommand.ConnectProperty property = new ConnectCommand.ConnectProperty();
        connectCommand.setConnectProperty(property);

        //字符串 	客户端连接到的服务器端应用的名字
        property.setApp("livetv");
        //字符串 	Flash Player 版本号。和ApplicationScript getversion() 方法返回的是同一个字符串。
        //property.setFlashver("LNX 9,0,124,2");
        List<String> versions = Lists.newArrayList();
        for (byte b : Constant.client_version) {
            versions.add(String.valueOf(b));
        }
        String join = StringUtils.join(versions, ",");
        property.setFlashver("LNX "+join);
        System.out.println(property.getFlashver());

        //字符串   进行当前连接的 SWF 文件源地址    	file://C:/FlvPlayer.swf
        property.setSwfUrl(null);
        property.setCapabilities(15.0);
        //字符串 	服务器 URL
        String pageUrl = StringUtils.substringBeforeLast(Utils.rtmp_url, "/");
        property.setPageUrl(pageUrl);
        //布尔    如果使用了代理就是 true
        property.setFpad(false);
        //数字    表明支持的视频编码
        property.setAudioCodecs(3575.0);
        //数字    表明所支持的特殊视频方法。
        property.setVideoFunction(1.0);
        property.setVideoCodecs(252.0);
        //字符串 	SWF 文件所加载的网页 URL。可以是undefined
        //property.setPageUrl("undefined");
        //数字    AMF 编码方法。0、3.一般采用3
        property.setObjectEncoding(0.0);

        MessageHeader messageHeader = new MessageHeader();
        messageHeader.setMessageStreamId(getMessageStreamId(session));
        messageHeader.setMessageType(Constant.message_type_20_command_message_amf0);
        messageHeader.setTimestamp(getTimestamp(session));
        rtmpHeader.setMessageHeader(messageHeader);
        session.write(connectCommand);
    }

    @Override
    public void sessionClosed(IoSession iosession) throws Exception {
        super.sessionClosed(iosession);
        System.exit(1);
    }

    @Override
    public void sessionIdle(IoSession iosession, IdleStatus idlestatus)
            throws Exception {
        System.out.println("客户端会话休眠");
        super.sessionIdle(iosession, idlestatus);
    }

    public static void main(String[] args) throws InterruptedException {
        MinaTcpClient client = new MinaTcpClient();
        // 关闭会话，待所有线程处理结束后
        // client.connector.dispose(true);
    }
}
